package bx.httpserver;

import akka.actor.typed.ActorRef;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.javadsl.ActorContext;
import akka.http.javadsl.Http;
import akka.http.javadsl.ServerBinding;
import akka.http.javadsl.server.Route;
import bx.httpserver.actor.JobManagerActor;
import bx.httpserver.router.JobWorkerRoutes;
import com.typesafe.config.Config;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletionStage;

public class AkkaJobWorkerServer {

    public static void startWorkerServer(ActorContext<Void> context) {

        ActorRef<JobManagerActor.JobManagerActorMsg> jobManagerActorMsgActor =
                context.spawn(JobManagerActor.create(), "bx-job-JobManagerActor");
        JobWorkerRoutes userRoutes = new JobWorkerRoutes(context.getSystem(), jobManagerActorMsgActor);
        startHttpServer(userRoutes.jobWorkerRoutes(), context.getSystem());

    }


    private static void startHttpServer(Route route, ActorSystem<?> system) {
        Config serverConfig = system.settings().config().getConfig("bx-job-worker.server");
        String ipAddr = serverConfig.getString("ip-addr");
        int port = serverConfig.getInt("port");

        CompletionStage<ServerBinding> futureBinding =
                Http.get(system).newServerAt(ipAddr, port).bind(route);

        futureBinding.whenComplete((binding, exception) -> {
            if (binding != null) {
                InetSocketAddress address = binding.localAddress();
                system.log().info("bx-job-worker online at http://{}:{}/",
                        address.getHostString(),
                        address.getPort());
            } else {
                system.log().error("Failed to bind HTTP endpoint, terminating system", exception);
                system.terminate();
            }
        });
    }
}
